package com.jielihaofeng.springboot.kafka.test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

/**
 * @author Johnnie Wind
 * @description 消息生产者
 * @date 2021/3/3 14:23
 */
@RestController
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    // 简单生产者：发送消息
    @GetMapping("/kafka/normal/{message}")
    public void sendMessage(@PathVariable("message") String message) {
        kafkaTemplate.send("topic1", "data_increment_data.kafka.hdp-kafka.test_source.test_table.*.*.*", message);
    }

    // 简单生产者：发送wormhole消息
    @GetMapping("/kafka/wormhole/{topic}/{key}/{message}")
    public void sendWormholeMessage(@PathVariable("topic") String topic, @PathVariable("key") String key, @PathVariable("message") String message) {
        kafkaTemplate.send(topic, key, message);
    }

    // 简单生产者：发送wormhole消息
    @PostMapping("/kafka/wormhole")
    public String sendWormholeMessagePost(@RequestParam(name = "topic") String topic, @RequestParam(name = "key") String key, @RequestParam(name = "message") String message) {
        ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send(topic, key, message);
        return "发送成功";
    }

    // 带回调的生产者1：发送消息
    @GetMapping("/kafka/callBackOne/{message}")
    public void sendMessage2(@PathVariable("message") String message) {
        kafkaTemplate.send("topic_wzy", "data_increment_data.kafka.hdp-kafka.test_source.test_table.*.*.*", message)
                .addCallback(success -> {
                    //消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();
                    System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
                }, failure -> {
                    System.err.println("发送消息失败：" + failure.getMessage());
                });
    }

    // 带回调的生产者2：发送消息
    @GetMapping("/kafka/callBackTwo/{message}")
    public void sendMessage3(@PathVariable("message") String message) {
        kafkaTemplate.send("topic_wzy", "data_increment_data.kafka.hdp-kafka.test_source.test_table.*.*.*", message)
                .addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        System.err.println("发送消息失败：" + throwable.getMessage());
                    }

                    @Override
                    public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                        //消息发送到的topic
                        String topic = stringObjectSendResult.getRecordMetadata().topic();
                        // 消息发送到的分区
                        int partition = stringObjectSendResult.getRecordMetadata().partition();
                        // 消息在分区内的offset
                        long offset = stringObjectSendResult.getRecordMetadata().offset();
                        System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
                    }
                });
    }

    // 提交事务
    @GetMapping("/kafka/transaction")
    public void sendMessage7() {
        // 声明事务：后面报错信息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("topic_wzy", "test transaction");
            throw new RuntimeException("测试事务");
        });

        // 不声明事务：后面报错但前面已经发出去了
        kafkaTemplate.send("topic_wzy", "test no transaction");
        throw new RuntimeException("测试无事务");
    }
}
